package com.zhang.storm.uv;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

public class UvSumBolt extends BaseRichBolt {

    private OutputCollector collector;

    private Map<String, Integer> map = new HashMap<>();

    @Override
    public void prepare(Map map, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        String ip = input.getString(0);
        Integer num = input.getInteger(1);
        if (map.containsKey(ip)) {
            map.put(ip, map.get(ip) + num);
        } else {
            map.put(ip, num);
        }
        System.err.println("ip:" + ip + "\tnum:" + map.get(ip));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}
